package com.youxin.logprocess.controller;

import java.io.File;
import java.io.Serializable;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hdfs.server.namenode.status_jsp;

import com.alibaba.fastjson.JSON;
import com.google.common.reflect.TypeToken;
import com.google.gson.Gson;
import com.youxin.busfoundation.bean.FileBeatBean;
import com.youxin.busfoundation.bean.MonitorAlarmBean;
import com.youxin.busfoundation.bean.MonitorMaxTimesBean;
import com.youxin.busfoundation.common.RedisKeyGenerator;
import com.youxin.busfoundation.common.RedisStorage;
import com.youxin.busfoundation.config.MonitorBaseBean;
import com.youxin.busfoundation.config.MonitorNginxVisitConfig;
import com.youxin.foundation.db.redis.RedisClusterCommand;
import com.youxin.foundation.utils.ConvertUtils;
import com.youxin.foundation.utils.JsonUtils;
import com.youxin.foundation.utils.RegexUtils;
import com.youxin.foundation.utils.collections.CollectionsUtils;
import com.youxin.foundation.utils.thread.ScheduleThreadPool;
import com.youxin.foundation.utils.timer.TimerSchedule;
import com.youxin.logprocess.bean.HostVisitBean;
import com.youxin.logprocess.bean.MessageBean;
import com.youxin.logprocess.bean.NginxBean;
import com.youxin.logprocess.bean.StatTimeBean;
import com.youxin.logprocess.common.MonitorGlobalConfig;
import com.youxin.logprocess.common.MonitorRedisKeyGenerator;

import redis.clients.jedis.JedisCluster;

/**
 * Nginx监控控制器
 * 
 * @author yingde.cao
 *
 */
public class MonitorNginxController extends BaseMonitorController implements Serializable {

	private static final long serialVersionUID = 4575894675745L;
	private static MonitorNginxController controller = new MonitorNginxController();
	protected Map<String, List<HostVisitBean>> nginxVisitTimesMap = Collections.synchronizedMap(new HashMap<>(4));
	/**
	 * 访问次数
	 */
	private Map<String, List<HostVisitBean>> hostVisitTimesMap = Collections.synchronizedMap(new HashMap<>(MonitorGlobalConfig.NGINX_ALARM_COUNT));
	private MonitorMaxTimesBean nginxMaxTimes, ipMaxTimes;
	private MonitorNginxVisitConfig config;

	public MonitorNginxVisitConfig getConfig() {
		return config;
	}

	public void setConfig(MonitorNginxVisitConfig config) {
		this.config = config;
	}

	public static void initConfig(MonitorNginxVisitConfig config) {
		controller.loadConfig(config);
	}

	public void loadConfig(MonitorNginxVisitConfig config) {
		if (this.config != null) {
			return;
		}
		try {
			setConfig(config);
			setNginxMaxTimes(config.getAlarm().getMaxtimes());
			this.ipMaxTimes = config.getIp().getMaxtimes();
			
			ScheduleThreadPool.scheduleWithFixedDelay(new Runnable() {
				
				@Override
				public void run() {
					scheduleUpdateRedis();					
				}
			}, 4000, 1000);			
			
		} catch (Exception e) {
			logger.error("loadConfig异常", e);
		}
	}

	public MonitorMaxTimesBean getIpMaxTimes() {
		return ipMaxTimes;
	}

	public void setIpMaxTimes(MonitorMaxTimesBean ipMaxTimes) {
		this.ipMaxTimes = ipMaxTimes;
	}

	private String getRootRedisKey()
	{
		return this.getMonitorBean().getId()+"-visit";
	}
	
	/**
	 * 定时更新Redis中的数据
	 */
	private void scheduleUpdateRedis() {
		try {
			String monitorGroupId = getRootRedisKey();
			final String lockMonitorGroupKey = MonitorRedisKeyGenerator.lockKey(monitorGroupId);
			if (nginxVisitTimesMap.size() > 0 || hostVisitTimesMap.size() > 0) {
				RedisClusterCommand.instance().lockCommand(lockMonitorGroupKey, 6000, (redisCluster) -> {
					try {
						/**
						 * 获取当前监控的异常信息
						 */
						String tRedisKey = monitorGroupId;
						Map<String, String> monitorNginxMap = redisCluster.hgetAll(tRedisKey);
						updateRedisMap(redisCluster, monitorNginxMap, tRedisKey);
					} catch (Exception e) {
						logger.error("执行任务时异常:" + monitorGroupId, e);
					}
				});
			}
		} catch (Exception e) {
			logger.error("执行定时任务发现异常", e);
		}
	}
	
	private synchronized Map<String, List<HostVisitBean>> copyToNewMapClear(Map<String, List<HostVisitBean>> map)
	{
		Map<String, List<HostVisitBean>> tempMap=new HashMap<>(map);
		map.clear();
		return tempMap;
	}

	private void updateRedisMap(JedisCluster redis, Map<String, String> monitorNginxMap, String redisKey) {
		StatTimeBean statTimeBean = new StatTimeBean();
		statTimeBean.generateNginxStatTimes();		
		
		Map<String, List<HostVisitBean>> tempNginxVisitMap=copyToNewMapClear(nginxVisitTimesMap);
		Map<String, List<HostVisitBean>> tempUserVisitMap=copyToNewMapClear(hostVisitTimesMap);
		/**
		 * 当前nginx所有主机时间的key值
		 */
		Map<String, String> resultValues = new HashMap<>(10);
		/**
		 * 一分钟
		 */
		String nginxMinutesKey = statTimeBean.getCurrentMinuteTimeKey();
		reComputeNginxVisitTimes(monitorNginxMap, resultValues, nginxMinutesKey, nginxMaxTimes.getMinute(), "一分钟",tempNginxVisitMap);

		/**
		 * 2分钟
		 */
		nginxMinutesKey = statTimeBean.getTwoMinutesTimeKey();
		reComputeNginxVisitTimes(monitorNginxMap, resultValues, nginxMinutesKey, nginxMaxTimes.getTwominute(), "2分钟",tempNginxVisitMap);

		/**
		 * 5分钟
		 */
		nginxMinutesKey = statTimeBean.getFiveMinutesTimeKey();
		reComputeNginxVisitTimes(monitorNginxMap, resultValues, nginxMinutesKey, nginxMaxTimes.getFiveminute(), "5分钟",tempNginxVisitMap);

		/**
		 * 20分钟
		 */
		nginxMinutesKey = statTimeBean.getTwentyMinuteTimeKey();
		reComputeNginxVisitTimes(monitorNginxMap, resultValues, nginxMinutesKey, nginxMaxTimes.getTwenty(), "20分钟",tempNginxVisitMap);
		
		/**
		 * host一分钟的访问
		 */
		String hostVisitTimesKey = statTimeBean.getCurrentMinuteHostTimeKey();
		reComputeUserVisitTimes(monitorNginxMap, resultValues, hostVisitTimesKey, ipMaxTimes.getMinute(),"1分钟",tempUserVisitMap);
		/**
		 * host两分钟的访问
		 */
		hostVisitTimesKey = statTimeBean.getTwoMinutesHostTimeKey();
		reComputeUserVisitTimes(monitorNginxMap, resultValues, hostVisitTimesKey, ipMaxTimes.getTwominute(),"2分钟",tempUserVisitMap);

		/**
		 * host五分钟的访问
		 */
		hostVisitTimesKey = statTimeBean.getFiveMinuteHostTimeKey();
		reComputeUserVisitTimes(monitorNginxMap, resultValues, hostVisitTimesKey, ipMaxTimes.getFiveminute(),"5分钟",tempUserVisitMap);

		/**
		 * 二十分钟
		 */
		hostVisitTimesKey = statTimeBean.getTwentyMinuteHostTimeKey();
		reComputeUserVisitTimes(monitorNginxMap, resultValues, hostVisitTimesKey, ipMaxTimes.getTwenty(),"20分钟",tempUserVisitMap);

		redisOperate(redis, redisKey, monitorNginxMap, resultValues);
	}
	

	/**
	 * 更新Redis中的数据
	 * 
	 * @param redisCluster
	 * @param redisKey
	 * @param monitorNginxMap
	 * @param resultValues
	 */
	private void redisOperate(JedisCluster redisCluster, String redisKey, Map<String, String> monitorNginxMap,
			Map<String, String> resultValues) {
		redisCluster.hmset(redisKey, resultValues);
		/**
		 * 查找旧的统计数据,旧的统计数据需要删除
		 */
		List<String> delkeys = monitorNginxMap.keySet().stream().filter((item) -> {
			return !resultValues.containsKey(item);
		}).collect(Collectors.toList());
		if (delkeys.size() > 0) {
			String[] delkeyArray = delkeys.toArray(new String[] {});
			redisCluster.hdel(redisKey, delkeyArray);
		}
		
	}

	/**
	 * 拷贝主机访问次数,重新计算用户访问主机次数,并保存到Redis本地集合中
	 * 
	 * @param sourceMap
	 *            源map
	 * @param destinationMap
	 *            目标map
	 * @param key
	 *            键值
	 */
	@SuppressWarnings({ "serial" })
	private void reComputeUserVisitTimes(Map<String, String> sourceMap, Map<String, String> destinationMap,
			String key, long maxTimes,String desc,Map<String, List<HostVisitBean>> tempUserVisitMap) {
		List<MessageBean> alarmIpList = new LinkedList<>();
		
		synchronized (this) {
			String sourceValue = sourceMap.get(key);
			Map<String, Long> hostMap = null;
			try {
				if (StringUtils.isNotEmpty(sourceValue)) {
					hostMap = new Gson().fromJson(sourceValue, new TypeToken<HashMap<String, Long>>() {
					}.getType());
				}
			} catch (Exception e) {
				logger.error("redis key:" + key + ",value:"
						+ com.youxin.foundation.utils.StringUtils.isNullSetEmpty(sourceValue), e);
			}
			if (hostMap == null) {
				hostMap = new HashMap<>(10);
			}
			List<HostVisitBean> hostVisitBeans = tempUserVisitMap.get(key);
			if (hostVisitBeans != null && hostVisitBeans.size() > 0) {
				for (HostVisitBean item : hostVisitBeans) {
					CollectionsUtils.incrementLong(hostMap, item.getIp(), item.getTimes());
				}				
				/**
				 * 对访问主机进行排序
				 */
				List<Map.Entry<String, Long>> list = new ArrayList<>(hostMap.entrySet());
				Collections.sort(list, new Comparator<Map.Entry<String, Long>>() {
					@Override
					public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
						return (o2.getValue()).compareTo(o1.getValue());
					}
				});
				
				if (hostMap.size() > MonitorGlobalConfig.NGINX_ALARM_COUNT) {					
					for (int i = list.size() - 1; i >= MonitorGlobalConfig.NGINX_ALARM_COUNT; --i) {
						Map.Entry<String, Long> item = list.get(i);
						hostMap.remove(item.getKey());
					}					
				}
				for (int i = 0; i < Math.min(MonitorGlobalConfig.NGINX_ALARM_COUNT,list.size()); ++i) {
					Map.Entry<String, Long> item = list.get(i);
					if (item.getValue().longValue() >= maxTimes) {
						alarmIpList.add(new MessageBean(item.getKey(),item.getKey()+"用户在"+desc+"的访问次数为:"+item.getValue().longValue()));
					}
				}
			}
			String jsonString = JsonUtils.objectToJsonString(hostMap);
			destinationMap.put(key, jsonString);
		}

		if(alarmIpList.size()>0)
		{
			add(alarmIpList);
			for(MessageBean item:alarmIpList)
			{
				RedisStorage.saveNginxAlarmHost(RedisClusterCommand.instance().getRedisCluster(), item.getHosts());
			}
		}	
	}

	/**
	 * 从HashMap拷贝一个Key到另外一个HashMap中,重新计算Nginx主机被访问次数,并保存到Redis本地集合中
	 * 
	 * @param sourceMap
	 *            源map
	 * @param destinationMap
	 *            目标map
	 * @param key
	 *            key名称
	 * @param times
	 *            次数
	 */
	@SuppressWarnings("serial")
	private long reComputeNginxVisitTimes(Map<String, String> sourceMap, Map<String, String> destinationMap, String key,
			long maxTimes, String desc,Map<String, List<HostVisitBean>> tempNginxVisitMap) {
		String sourceValue = sourceMap.get(key);
		Map<String, Long> hostMap = null;
		try {
			if (StringUtils.isNotEmpty(sourceValue)) {
				hostMap = new Gson().fromJson(sourceValue, new TypeToken<HashMap<String, Long>>() {
				}.getType());
			}
		} catch (Exception e) {
			logger.error("redis key:" + key + ",value:"
					+ com.youxin.foundation.utils.StringUtils.isNullSetEmpty(sourceValue), e);
		}
		if (hostMap == null) {
			hostMap = new HashMap<>(10);
		}
		List<MessageBean> alarmHosts=new LinkedList<>();
		synchronized (this) {
			List<HostVisitBean> hostVisitBeans = tempNginxVisitMap.get(key);
			if (hostVisitBeans != null && hostVisitBeans.size() > 0) {
				for (HostVisitBean item : hostVisitBeans) {
				  long times=CollectionsUtils.incrementLong(hostMap, item.getIp(), item.getTimes());
				  if(times>=maxTimes)
				  {
					  alarmHosts.add(new MessageBean(item.getIp(), "服务器"+item.getIp()+"在"+desc+"内被访问的次数:"+times));					  
				  }
				}
			}
			destinationMap.put(key, JsonUtils.objectToJsonString(hostMap));
		}
		
		if(alarmHosts.size()>0)
		{
			add(alarmHosts);
		}
		
		return 0;
	}

	public MonitorMaxTimesBean getNginxMaxTimes() {
		return nginxMaxTimes;
	}

	public void setNginxMaxTimes(MonitorMaxTimesBean nginxMaxTimes) {
		this.nginxMaxTimes = nginxMaxTimes;
	}

	public static MonitorNginxController instance() {
		return controller;
	}

	@Override
	public MonitorBaseBean getMonitorBean() {
		return config;
	}

	/**
	 * Spark 调用
	 * 
	 * @param fileBeatBean
	 */
	public void execute(FileBeatBean fileBeatBean) {
		try {
			if(config.getMonitor()==0){
				return;
			}
			String message = fileBeatBean.getMessage();
			String[] regexs = new String[] { "(\\d{1,3}(?:\\.\\d{1,3}){3})", "([\\w\\-]+)", "([^\\[\\]]+)", "([^\"]+)",
					"(\\d+)", "(\\d+)", "([^\"]+)", "([^\"]+)", "([^\"]+)" };
			String regex = regexs[0] + "\\s+\\-\\s+" + regexs[1] + "\\s+\\[" + regexs[2] + "\\]\\s+\"" + regexs[3]
					+ "\"\\s+" + regexs[4] + "\\s+" + regexs[5] + "\\s+\"" + regexs[6] + "\"\\s+\"" + regexs[7]
					+ "\"\\s+\"" + regexs[8] + "\"\\s*";

			NginxBean nginxBean = RegexUtils.match(regex, message, Pattern.MULTILINE, (matcher, success) -> {
				if (success) {
					NginxBean bean = new NginxBean();
					bean.setHost(fileBeatBean.getBeat().getHostname());
					String remoteAddr = matcher.group(1);
					bean.setRemoteAddr(remoteAddr);
					String timeLocal = matcher.group(3);
					bean.setTimeLocal(timeLocal);
					String request = matcher.group(4);
					bean.setRequest(request);					
					String httpReferer = matcher.group(7);
					bean.setHttpReferer(httpReferer);
					String httpXforwardedFor = matcher.group(9);
					bean.setHttpXforwardedFor(httpXforwardedFor);
					
					return bean;
				}
				return null;
			});
			/**
			 * 判断是否获取到时间,是否获取到GET/POST请求标识
			 */
			if (nginxBean != null 
					&& nginxBean.getTimeLocalDate() != null 
					&& nginxBean.getRequestBean() != null) {
				if(nginxBean.getRequestBean().isSupportStatics())
				{
					updateNginxVisitTimes(nginxBean);
				}
			}

		} catch (Exception e) {
			logger.error("转换错误", e);
		}
	}

	/**
	 * 更新本地HashMap,当前一分钟内 minute:201710311000=>times
	 * 
	 * @param nginxBean
	 */
	private void updateNginxVisitTimes(NginxBean nginxBean) {
		StatTimeBean statTimeBean = new StatTimeBean();
		statTimeBean.generateNginxStatTimes();
		/**
		 * 如果时间不在20分钟内,就不处理这些旧的日志
		 */
		if (!statTimeBean.isInTwentyMinutesTime(nginxBean.getTimeLocalDate())) {
			return;
		}
		String remoteAddr = nginxBean.getRemoteAddr();
		String nginxHost = nginxBean.getHost();
		/**
		 * 20分钟内
		 */
		incrementNginxVisitTimes(nginxHost, statTimeBean.getTwentyMinuteTimeKey());
		incrementVisitTimes(remoteAddr, statTimeBean.getTwentyMinuteHostTimeKey());

		if (!statTimeBean.isInFiveMinutesTime(nginxBean.getTimeLocalDate())) {
			return;
		}

		/**
		 * 5分钟内
		 */
		incrementNginxVisitTimes(nginxHost, statTimeBean.getFiveMinutesTimeKey());
		incrementVisitTimes(remoteAddr, statTimeBean.getFiveMinuteHostTimeKey());

		if (!statTimeBean.isInTwoMinutesTime(nginxBean.getTimeLocalDate())) {
			return;
		}
		/**
		 * 两分钟内
		 */
		incrementNginxVisitTimes(nginxHost, statTimeBean.getTwoMinutesTimeKey());
		incrementVisitTimes(remoteAddr, statTimeBean.getTwoMinutesHostTimeKey());

		if (!statTimeBean.isInCurrentMinuteTime(nginxBean.getTimeLocalDate())) {
			return;
		}
		/**
		 * 一分钟内
		 */
		incrementNginxVisitTimes(nginxHost, statTimeBean.getCurrentMinuteTimeKey());
		incrementVisitTimes(remoteAddr, statTimeBean.getCurrentMinuteHostTimeKey());
	}

	/**
	 * 增加当前Nginx主机的访问次数
	 * 
	 * @param hostName
	 *            主机名称
	 * @param timeKey
	 *            key名称,时间点名
	 */
	private synchronized void incrementNginxVisitTimes(String hostName, String timeKey) {
		List<HostVisitBean> hostVisitBeans = nginxVisitTimesMap.get(timeKey);
		if (hostVisitBeans == null) {
			hostVisitBeans = new ArrayList<>(3);
			hostVisitBeans.add(new HostVisitBean(hostName, 1));
			nginxVisitTimesMap.put(timeKey, hostVisitBeans);
		} else {
			HostVisitBean hostVisitBean = findHostByIp(hostName, hostVisitBeans);
			if (hostVisitBean != null) {
				hostVisitBean.incrementTimes();
			} else {
				hostVisitBeans.add(new HostVisitBean(hostName, 1));
			}
		}
	}

	/**
	 * 增长ip的访问次数
	 * 
	 * @param remoteAddr
	 *            主机ip
	 */
	private synchronized void incrementVisitTimes(String remoteAddr, String timeKey) {
		List<HostVisitBean> hostVisitBeans = hostVisitTimesMap.get(timeKey);
		if (hostVisitBeans == null) {
			hostVisitBeans = new ArrayList<>(MonitorGlobalConfig.NGINX_ALARM_COUNT);
			hostVisitBeans.add(new HostVisitBean(remoteAddr, 1));
			hostVisitTimesMap.put(timeKey, hostVisitBeans);
		} else {
			HostVisitBean hostVisitBean = findHostByIp(remoteAddr, hostVisitBeans);
			if (hostVisitBean != null) {
				hostVisitBean.incrementTimes();
			} else {
				if (hostVisitBeans.size() < MonitorGlobalConfig.NGINX_ALARM_COUNT) {
					hostVisitBeans.add(new HostVisitBean(remoteAddr, 1));
				}
			}
		}
	}

	/**
	 * 查找主机
	 * 
	 * @param remoteAddr
	 *            主机地址
	 * @param hostVisitBeans
	 *            主机地址
	 * @return HostVisitBean
	 */
	private HostVisitBean findHostByIp(String remoteAddr, List<HostVisitBean> hostVisitBeans) {
		for (HostVisitBean item : hostVisitBeans) {
			if (item.getIp().equals(remoteAddr)) {
				return item;
			}
		}
		return null;
	}

}
